package com.wlz.rocketmq.simple;

import com.wlz.rocketmq.constant.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.CountDownLatch;

/**
 *  异步发送消息
 * @author wlz
 * @date 2022-07-01  9:18 下午
 */
@Slf4j
public class AsyncProducer {

    public static void main(String[] args) throws Exception{

        DefaultMQProducer producer = new DefaultMQProducer("AsyncProducer");

        producer.setNamesrvAddr(Constants.NAMESVADDR);

        producer.start();

        // 重试次数
        producer.setRetryTimesWhenSendAsyncFailed(0);

        int messageCount = 100;

        //由于是异步发送，这里引入一个countDownLatch，保证所有Producer发送消息的回调方法都执行完了再停止Producer服务。
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);

        for (int i = 0; i < messageCount; i++) {
            try {
                final int index = i;
                Message msg = new Message("TopicTest", "TagA", "OrderID188",
                        "hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        log.info("%-10d OK {}, {}", index, sendResult.getMsgId());
                    }

                    @Override
                    public void onException(Throwable e) {
                        countDownLatch.countDown();
                        log.info("%-10d Exception {}, {}", index, e);
                    }
                });
                log.info("消息发送完成");
            }catch (Exception e) {
                log.info("error: {}",e);
            }
        }
        countDownLatch.await();
        producer.shutdown();

    }
}
